

#include "TcpConnection.hpp"
#include "Channel.hpp"
#include "Socket.hpp"
#include "Logger.hpp"
#include "EventLoop.hpp"


namespace zephyr {

TcpConnection::TcpConnection(EventLoop *loop, const std::string &nameArg,
                             int sockfd, const InetAddress &localaddr, const InetAddress &peeraddr)
    : m_loop(loop)
    , m_name(nameArg)
    , m_state(Connecteing)
    , m_reading(true)
    , m_socket(new Socket(sockfd))
    , m_channel(new Channel(loop, sockfd))
    , m_localaddr(localaddr)
    , m_peeraddr(peeraddr)
    , m_highWaterMark(64 * 1024 * 1024) {

    // 给channel设置对应的回调函数
    // poller通知channel感兴趣的事件发生
    // channel就会调用对应的回调函数
    m_channel->setReadCallback(
        std::bind(&TcpConnection::handleRead, this, std::placeholders::_1)
    );

    m_channel->setWriteCallback(
        std::bind(&TcpConnection::handleWrite, this)
    );

    m_channel->setCloseCallback(
        std::bind(&TcpConnection::handleClose, this)
    );

    m_channel->setErrorCallback(
        std::bind(&TcpConnection::handleError, this)
    );

    LOG_INFO("TcpConnection::ctor[%s] as fd = %d\n", m_name.c_str(), sockfd);
    m_socket->setKeepAlive(true);
}

TcpConnection::~TcpConnection() {
    LOG_INFO("TcpConnection::dtor[%s] at fd = %d state = %d\n", m_name.c_str(), m_channel->fd(), (int)m_state);
}

void TcpConnection::send(const std::string &buf) {
    if (m_state == Connected) {
        if (m_loop->isInLoopThread()) { // 单reactor情况 调用send时loop即为main_loop
            sendInLoop(buf.c_str(), buf.size());
        } else {
            m_loop->runInLoop(
                std::bind(&TcpConnection::sendInLoop, this, buf.c_str(), buf.size())
            );
        }
    }
}

void TcpConnection::sendInLoop(const void *data, size_t len) {
    ssize_t nwrote = 0;
    size_t remaining = len;

    bool faultError = false;

    if (m_state == Disconnected) {
        LOG_ERROR("disconnected, give up writing");
    }

    // 表示channel第一次写数据 或缓冲区没有待发送数据
    if (!m_channel->isWriting() && m_outputBuffer.readableBytes() == 0) {
        nwrote = ::write(m_channel->fd(), data, len);
        if (nwrote >= 0) {
            remaining = len = nwrote;
            if (remaining == 0 && m_writeCompletCallback) {
                // 数据全部发送完成，不用再给channel设置epollout事件了
                m_loop->queueInLoop(
                    std::bind(m_writeCompletCallback, shared_from_this())
                );
            }
        } else {
            nwrote = 0;
            if (errno != EWOULDBLOCK) {
                LOG_ERROR("TcpConnection::sendInLoop");
                if (errno == EPIPE || errno == ECONNRESET) {
                    faultError = true;
                }
            }
        }
    }

    /**
     * 还没有将所有的数据发送出去
     * channel的writeCallback_实际上就是TcpConnection设置的handleWrite回调
     */
    if (!faultError && remaining > 0) {
        // 当前发送缓冲区的待发送数据的长度
        size_t oldLen = m_outputBuffer.readableBytes();
        if (oldLen + remaining >= m_highWaterMark && oldLen < m_highWaterMark && m_highWaterMarkCallback) {
            m_loop->queueInLoop(
                std::bind(m_highWaterMarkCallback, shared_from_this(), oldLen + remaining)
            );
        }
        m_outputBuffer.append((char *)data + nwrote, remaining); // 将剩下的数据写入发送缓冲区
        if (!m_channel->isWriting()) {
            m_channel->enableWriting(); // 向loop中的poll 注册可写事件
        }
    }
}

void TcpConnection::shutdown() {
    if (m_state == Connected) {
        setState(Disconnecting);
        m_loop->runInLoop(
            std::bind(&TcpConnection::shutdownInLoop, this)
        );
    }
}

void TcpConnection::shutdownInLoop() {
    if (!m_channel->isWriting()) { // 说明当前outputbuffer中所有数据已经发送完成
        m_socket->shutdownWrite();
    }
}

// 连接建立
void TcpConnection::connectEstablished() {
    setState(Connected);
    m_channel->tie(shared_from_this());
    m_channel->enableReading(); // 注册可读事件
    // 连接建立执行回调
    m_connectionCallback(shared_from_this());
}

void TcpConnection::connectDestroyed() {
    if (m_state == Connected) {
        setState(Disconnected);
        m_channel->disableAll();
        m_connectionCallback(shared_from_this());
    }
    m_channel->remove(); // 将channel从poller中删除
}

// 读是相对服务器而言的 当对端客户端有数据到达 服务器端检测到EPOLLIN 就会触发该fd上的回调 handleRead取读走对端发来的数据
void TcpConnection::handleRead(Timestamp recviveTime) {
    int saveErrno = 0;
    ssize_t n = m_inputBuffer.readFd(m_channel->fd(), &saveErrno);
    if (n > 0) { // 有数据到达，执行数据到达回调
        m_messageCallback(shared_from_this(), &m_inputBuffer, recviveTime);
    } else if (n == 0) { // 客户端断开连接
        handleClose();
    } else {
        errno = saveErrno;
        LOG_ERROR("TcpConnection::handleRead");
        handleError();
    }
}

void TcpConnection::handleWrite() {
    if (m_channel->isWriting()) {
        int saveErrno = 0;
        ssize_t n = m_outputBuffer.writeFd(m_channel->fd(), &saveErrno);
        if (n > 0) {
            // 将写缓冲区中已写的数据抛弃
            m_outputBuffer.retrieve(n);
            if (m_outputBuffer.readableBytes() == 0) { // 没有可写数据
                m_channel->disableWriting(); // 关闭可写事件
                if (m_writeCompletCallback) { // 如果设置写完成回调，执行回调
                    m_loop->queueInLoop(
                        std::bind(m_writeCompletCallback, shared_from_this())
                    );
                }
                if (m_state == Disconnecting) {
                    shutdownInLoop();
                }
            }
        } else {
            LOG_ERROR("TcpConnection::handleWrite");
        }
    } else {
        LOG_ERROR("TcpConnectionf fd=%d is down, no more writing", m_channel->fd());
    }
}

void TcpConnection::handleClose() {
    LOG_INFO("TcpConnection::handleClose fd=%d, state=%d\n", m_channel->fd(), (int)m_state);
    setState(Disconnecting);
    m_channel->disableAll();

    TcpConnectionPtr connPtr(shared_from_this());
    m_connectionCallback(connPtr); // 连接回调
    m_closeCallback(connPtr);  // 执行关闭连接的回调 执行的是TcpServer::removeConnection方法
}

void TcpConnection::handleError() {
    int optval;
    socklen_t optlen = sizeof(optval);
    int err = 0;
    if (::getsockopt(m_channel->fd(), SOL_SOCKET, SO_ERROR, &optval, &optlen) < 0) {
        err = errno;
    } else {
        err =  optval;
    }
    LOG_ERROR("TcpConnection::handleError name:%s - SO_ERROR:%d\n", m_name.c_str(), err);
}


}